import datetime
import json
import math
import traceback
import queue
import threading
import time
import one_runtime
import script
import util
from script.common_client import DataSelector
from script.common_storage import ScriptNodeData, node_data_storage, eval_data_storage, call_record_storage, plan_storage, ScriptEvalData
from script import NodeResult, NodeTaskCtx, common_storage, script_thread_local


# 创建任务
def create_node_task(name: str = None, script_name=None, node=None, data_group=None, source: dict = None, wait: bool = True, pre_check: bool = False):
    source = source if source else {}
    source_group = source.get('source_group')
    source_node = source.get('source_node')
    source_type = source.get('source_type')
    offset = source.get('offset')
    size = source.get('size')
    search = source.get('search')
    begin_time = source.get('begin_time')
    end_time = source.get('end_time')
    spread_nodes = source.get('spread_nodes')

    if not script_name or not node or not data_group or not source_group or not name or not source_node:
        raise ValueError("参数不全")
    data_group = data_group.strip()
    if ',' in data_group or '，' in data_group or '|' in data_group:
        raise ValueError("产出数据集不可存在逗号或者|")
    curr_script = script.get_script_schemas().get(script_name)
    if not curr_script:
        raise ValueError('脚本不存在')
    functions = curr_script['functions']

    node_func = functions.get(node)
    if not node_func:
        raise ValueError('节点不存在')
    if spread_nodes:
        def bexistsa(anode, bnode):
            if anode == bnode:
                return True
            funa = functions.get(anode)
            for next_node in funa.get('next_nodes'):
                if bexistsa(next_node, bnode):
                    return True
            return False

        for spread_node in spread_nodes.split(','):
            if spread_node == node:
                raise ValueError('不可以扩散自己')
            if not functions.get(spread_node):
                raise ValueError('扩散节点不存在')
            if not bexistsa(node, spread_node):
                raise ValueError('扩散节点不在链路内')

    # 判断是否是根节点放行，放开限制可方便进行数据导入
    if source_node == script_name:
        pass  # if not node_func.get('root'):  #     raise ValueError('节点非根节点，必须指定上游节点')
    else:
        if source_node not in functions:
            raise ValueError('来源节点不存在')
        if node not in functions.get(source_node).get('next_nodes'):
            raise ValueError('节点不是来源节点的下游')
    # type过滤
    if node_func.get('accept_types'):
        if source_type:
            real_types = set(source_type.split(',')) & set(node_func.get('accept_types'))
        else:
            real_types = node_func.get('accept_types')
        # 没有就随便给个不可能存在的
        if not real_types:
            real_types = ['cbjhoy767uycdbhcdb87']
        source_type = ','.join(real_types)
        source.update({
            'source_type': source_type
        })

    # 相同任务
    old_task = one_runtime.tasks.get(name)
    if old_task:
        raise ValueError('当前任务正在执行中,请勿重复创建')

    task = NodeTask(name=name, script_name=script_name, node=node, data_group=data_group, source=source)
    if pre_check != True:
        util.log('create_task', {
            'task_name': task.name,
            'task': task
        })
        one_runtime.tasks.update({
            name: task
        })

    datas = node_data_storage.select_task_datas(script_name, node, data_group, source_node, source_group, source_type, begin_time, end_time, search, offset, size)
    # # 查询上游数据
    # datas = node_data_storage.select_datas(filters={
    #     'script_name': script_name,
    #     'node': source_node,
    #     'data_group': source_group,
    #     'content_type': source_type,
    #     'content_not_null': True,
    #     'search': search,
    #     'begin_time': begin_time,
    #     'end_time': end_time
    # }, offset=offset, size=size)
    # # 在过滤已经有数据的
    # if datas:
    #     # 查出来当前节点已消费的上游id列表
    #     complete_datas = node_data_storage.select_datas(filters={
    #         'source_ids': [item.id for item in datas if item.id],
    #         'node': node,
    #         'script_name': script_name,
    #         'data_group': data_group,
    #         'source_node': source_node
    #     })
    #     # 仅保留未执行部分
    #     # 先将 complete_datas 中的 source_id 提取到一个集合中
    #     completed_source_ids = {item_b.source_id for item_b in complete_datas}
    #
    #     # 仅保留未执行部分
    #     datas = [item_a for item_a in datas if item_a.id not in completed_source_ids]

    if pre_check == True:
        task.num = len(datas)
        return task
    if wait:
        # 初始化worker
        task.start_workers(datas=datas)
    else:
        # 后台执行
        thread = threading.Thread(target=task.start_workers, args=(datas,))
        thread.start()
    return task


# 创建评测任务
def create_eval_task(name: str = None, script_name: str = None, eval_node: str = None, eval_name: str = None, source: dict = None, wait: bool = True, pre_check: bool = False):
    source = source if source else {}
    source_node = source.get('source_node')
    source_group = source.get('source_group')
    data_groups = source.get('data_groups')

    if not script_name or not eval_node or not data_groups or not source_group or not name or not eval_name or not name or not source_node:
        raise ValueError("参数不全")

    curr_script = script.get_script_schemas().get(script_name)
    if not curr_script:
        raise ValueError('脚本不存在')
    functions = curr_script['functions']

    node_func = functions.get(eval_node)
    if not node_func:
        raise ValueError('节点不存在')

    eval_func = node_func.get('eval_func')
    if not eval_func:
        raise ValueError('节点未配置评测函数')
    # 判断是否是根节点放行，放开限制可方便进行数据导入
    if source_node == script_name:
        pass  # if not node_func.get('root'):  #     raise ValueError('节点非根节点，必须指定上游节点')
    else:
        if source_node not in functions:
            raise ValueError('来源节点不存在')
        if eval_node not in functions.get(source_node).get('next_nodes'):
            raise ValueError('节点不是来源节点的下游')

    # 相同任务
    old_task = one_runtime.eval_tasks.get(name)
    if old_task:
        raise ValueError('当前评测任务正在执行中,请勿重复创建')

    task = EvalTask(name=name, script_name=script_name, eval_node=eval_node, eval_name=eval_name, source=source)
    if pre_check != True:
        util.log('create_eval_task', {
            'task_name': task.name,
            'task': task
        })
        one_runtime.eval_tasks.update({
            name: task
        })

    # 查询上游数据
    datas = eval_data_storage.select_need_eval_datas(filters={
        'script_name': script_name,
        'source_node': source_node,
        'source_group': source_group,
        'eval_node': eval_node,
        'data_groups': data_groups,
        'eval_name': eval_name
    })

    if pre_check == True:
        task.num = len(datas)
        return task
    if wait:
        # 初始化worker
        task.start_workers(datas=datas)
    else:
        # 后台执行
        thread = threading.Thread(target=task.start_workers, args=(datas,))
        thread.start()
    return task

class NodeTask():
    def __init__(self, name: str = None, script_name: str = None, data_group: str = None, node: str = None, source: dict = None):
        self.task_type = 'node_task'
        self.name = name
        self.script_name = script_name
        self.data_group = data_group
        self.node = node
        self.status = None
        self.source = source
        self.monitor = {
            'details': [],
            'count_window': 300
        }
        self.queue = queue.Queue()
        self.num = 0
        self.workers = []
        self.success_num = 0
        self.filter_num = 0
        self.fail_num = 0
        self.money = 0
        self.prepare_result = {}

    def start_workers(self, datas: list[ScriptNodeData]):
        self.num = len(datas)
        self.status = 'on'
        if self.num > 0:
            # 初始化队列
            for data in datas:
                if data is not None:
                    self.queue.put(data)

            curr_script = script.get_script_schemas().get(self.script_name)
            if not curr_script:
                raise ValueError('脚本不存在')
            module = script.get_script_modules().get(self.script_name)
            if not module:
                #     脚本不存在，可能是被删除了或者改名了。
                raise ValueError('module不存在，可能被删除了')
            # Get the function from the module
            functions = curr_script['functions']
            node_func = functions.get(self.node)
            if not node_func:
                raise ValueError('节点不存在')
            prepare_func = node_func.get('prepare_func')
            if prepare_func:
                try:
                    plan_config = self.source.get('plan_config')
                    test_name = self.source.get('test_name')
                    debug_param = self.source.get('debug_param')
                    pass_through = self.source.get('pass_through')
                    ctx = NodeTaskCtx(task_name=self.name, task_source=self.source, script_name=self.script_name, node=self.node, data_group=self.data_group, req_id=None, plan_config=plan_config, test_name=test_name, debug_param=debug_param, pass_through=pass_through)
                    self.prepare_result = getattr(module, prepare_func)(datas, ctx)
                except:
                    self.status = 'off'
                    util.tell_dingding({
                        "msgtype": "text",
                        "text": {
                            "content": f'''脚本熔断告警\n脚本：{self.script_name}\n节点：{self.node}\n任务：{self.name}\nprepare_func执行失败\n即便重启任务也无法重新执行prepare_func。\n自动暂停'''
                        }
                    })

            worker_num = self.source.get('worker_num')
            if not worker_num:
                # 推荐size
                worker_num = node_func.get('suggest_worker_num')
                worker_num = 1 if not worker_num or worker_num < 1 else worker_num
            self.update_workers(worker_num)
            # 等待所有任务完成
            self.queue.join()
        self.clean_on_complete()

    def update_workers(self, worker_num):
        worker_num = worker_num if worker_num > 0 else 1
        current_worker_num = len(self.workers)
        util.log('update_workers', {
            'old_num': current_worker_num,
            'new_num': worker_num,
            'task_name': self.name,
            'script_name': self.script_name,
            'node': self.node
        })
        if current_worker_num < worker_num:
            # 添加新的worker
            for i in range(current_worker_num, worker_num):
                worker = NodeTask.WorkerThread(self, i)
                worker.start()
                self.workers.append(worker)
        elif current_worker_num > worker_num:
            # 停止多余的worker
            for i in range(worker_num, current_worker_num):
                self.workers[i].stop()
            self.workers = self.workers[:worker_num]

    # 并行回调，更新任务为complete
    def clean_on_complete(self):
        self.status = 'complete'
        util.log('complete_task', {
            'task_name': self.name
        }, '准备清理数据')
        node_data_storage.clean_repeats()
        # 删除无效调用关联
        call_record_storage.clean_records()
        util.log('complete_task', {
            'task_name': self.name
        }, '清理数据完毕')

        # 是否需要扩散
        spread_nodes = self.source.get('spread_nodes')
        if not spread_nodes:
            return
        curr_script = script.get_script_schemas().get(self.script_name)
        if not curr_script:
            util.log('spread_task_error', {
                'task_name': self.name,
                'script_name': self.script_name
            }, '脚本被删除')
            return
        functions = curr_script['functions']
        node_func = functions.get(self.node)
        if not node_func:
            util.log('spread_task_error', {
                'task_name': self.name,
                'script_name': self.script_name,
                'node': self.node
            }, '节点被删除')
            return

        def bexistsa(anode, bnode):
            if anode == bnode:
                return True
            funa = functions.get(anode)
            for next_node in funa.get('next_nodes'):
                if bexistsa(next_node, bnode):
                    return True
            return False

        for next_node in node_func.get('next_nodes'):
            try:
                source_spread_nodes = spread_nodes.split(',')
                if next_node not in source_spread_nodes:
                    continue
                #     删除自己并保留自己的下游
                source_spread_nodes.remove(next_node)
                source_spread_nodes = [item for item in source_spread_nodes if bexistsa(next_node, item)]

                # 使用默认方案
                plan = plan_storage.select_online_plan(self.script_name, next_node)
                plan_config = plan.config if plan else {}
                new_source = {
                    'spread_nodes': ','.join(source_spread_nodes),
                    'source_group': self.data_group,
                    'source_node': self.node,
                    'test_name': self.source.get('test_name'),
                    'pass_through': self.source.get('pass_through'),
                    'plan_config': plan_config
                }
                create_node_task(name=self.name + '#' + next_node, script_name=self.script_name, node=next_node, data_group=self.data_group, source=new_source, wait=False, pre_check=False)
                util.log('spread_task_success', {
                    'source': self.to_json(),
                    'next_node': next_node
                })
            except Exception as e:
                util.log('spread_task_error', {
                    'error': e,
                    'source': self.to_json(),
                    'next_node': next_node
                })

    def to_dict(self):
        return {
            "name": self.name,
            'task_type': self.task_type,
            "script_name": self.script_name,
            "data_group": self.data_group,
            "node": self.node,
            "worker_num": len(self.workers),
            "num": self.num,
            "success_num": self.success_num,
            "filter_num": self.filter_num,
            "fail_num": self.fail_num,
            "money": self.money,
            "status": self.status,
            "source": self.source,
            "monitor": self.monitor
        }

    def to_json(self):
        return json.dumps(self.to_dict(), ensure_ascii=False, default=str)

    def __repr__(self):
        return self.to_json()

    class WorkerThread(threading.Thread):
        # 工作线程
        def __init__(self, task, index):
            super(NodeTask.WorkerThread, self).__init__()
            self.task = task
            self.index = index
            self.is_stopped = False

        def run(self):
            util.log('task_worker_start', {
                'task_name': self.task.name,
                'script_name': self.task.script_name,
                'node': self.task.node,
                'worker_index': self.index
            })
            plan_config = self.task.source.get('plan_config')
            test_name = self.task.source.get('test_name')
            debug_param = self.task.source.get('debug_param')
            pass_through = self.task.source.get('pass_through')
            time.sleep(5 + self.index)  # 暂停5秒
            monitor_info = self.task.monitor
            index = 0
            while True:
                if self.is_stopped:
                    util.log('worker_stop', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'node': self.task.node,
                        'worker_index': self.index
                    })
                    break
                try:
                    data = self.task.queue.get()
                except queue.Empty:
                    util.log('queue_empty', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'node': self.task.node,
                        'worker_index': self.index
                    })
                    break
                if not data:
                    continue
                req_id = None
                start_time = time.time()
                try:
                    index += 1
                    if self.task.status == 'delete':
                        util.log('judge_status', {
                            'task_name': self.task.name,
                            'script_name': self.task.script_name,
                            'node': self.task.node,
                            'worker_index': self.index
                        }, '任务删除，跳出执行')
                        break

                    while self.task.status == 'off':
                        util.log('judge_status', {
                            'task_name': self.task.name,
                            'script_name': self.task.script_name,
                            'node': self.task.node,
                            'worker_index': self.index
                        }, '任务暂停，等待执行')
                        time.sleep(10)

                    # 记录开始时间
                    start_time = time.time()
                    # 重新查一下数据
                    data = node_data_storage.select_task_data_by_id(data.id, self.task.script_name, self.task.node, self.task.data_group)
                    if not data or data.content is None or (isinstance(data.content, str) and len(data.content) == 0):
                        self.task.filter_num += 1
                        util.log('filter_task', {
                            'task_name': self.task.name,
                            'script_name': self.task.script_name,
                            'node': self.task.node,
                            'do_index': str(index),
                            'worker_index': self.index,
                            'info': data
                        }, '跳过数据')
                        continue

                    req_id = str(time.time()) + '.' + str(self.index)
                    util.log('start_task', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'node': self.task.node,
                        'do_index': str(index),
                        'worker_index': self.index,
                        'info': data
                    }, '处理数据')
                    # 记录线程变量
                    script_thread_local.run_node_data = {
                        'task': self.task,
                        'script_name': self.task.script_name,
                        'node': self.task.node,
                        'data_group': self.task.data_group,
                        'worker_num': len(self.task.workers),
                        'data': data,
                        'req_id': req_id,
                        'test_name': test_name,
                        'debug_param': debug_param
                    }
                    util.log_local.run_data = {
                        'run_task_id': req_id,
                        'test_name': test_name,
                        'debug_param': debug_param
                    }
                    ctx = NodeTaskCtx(task_name=self.task.name, task_source=self.task.source, script_name=self.task.script_name, node=self.task.node, data_group=self.task.data_group, req_id=req_id, plan_config=plan_config, test_name=test_name, debug_param=debug_param, pass_through=pass_through,
                                      prepare_result=self.task.prepare_result)
                    node_datas = self.__run_node(data, ctx)
                    common_storage.node_data_storage.add_datas(node_datas)
                    self.task.success_num += 1
                    cost = math.ceil((time.time() - start_time) * 1000)
                    monitor_info['details'].append((start_time, cost, None, req_id))
                    # 10s统计一次
                    if monitor_info.get('last_count_time') is None or start_time - monitor_info.get('last_count_time') >= 10:
                        self.__count_task_monitor(start_time)
                except Exception as e:
                    self.task.fail_num += 1
                    monitor_info['details'].append((start_time, -1, traceback.format_exc(), req_id))
                    # 一旦失败就计算
                    self.__count_task_monitor(start_time)
                    util.log('task_error', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'node': self.task.node,
                        'data': data,
                        'error': e
                    }, '节点处理异常')
                finally:
                    # 清理线程局部变量
                    if hasattr(script_thread_local, "run_node_data"):
                        del script_thread_local.run_node_data
                    # 清理线程局部变量
                    if hasattr(util.log_local, "run_data"):
                        del util.log_local.run_data
                    self.task.queue.task_done()

        def stop(self):
            self.is_stopped = True

        def __run_node(self, data: ScriptNodeData, ctx: NodeTaskCtx = None):
            # script_name 是py脚本名称，该脚本在scripts目录下
            # node 是py脚本下的函数名称 该node有注解snode
            # 根据 script_name 和 node 获取指定函数，并将data传入 执行他，并将它的返回结果返回
            module = script.get_script_modules().get(self.task.script_name)
            if not module:
                #     脚本不存在，可能是被删除了或者改名了。
                raise ValueError('module不存在，可能被删除了')
            # Get the function from the module
            function = getattr(module, self.task.node)
            # Call the function with the provided data and return the result
            next_datas = function(data, ctx)

            # 构建节点信息
            source_node = data.node
            source_id = data.id
            source_nodes = {
                source_node: source_id
            }
            if data.source_nodes:
                source_nodes.update(data.source_nodes)
            gmt_create = datetime.datetime.now()
            next_datas = [NodeResult()] if not next_datas else next_datas if isinstance(next_datas, (list, set, tuple)) else [next_datas]

            # 存储执行细节
            def merge_ext(ext_info):
                if ext_info is None:
                    ext_info = {}
                elif not isinstance(ext_info, dict):
                    raise ValueError('ext_info must be dict')
                ext_info.update({
                    "pass_through": ctx.pass_through,
                    "plan_config": ctx.plan_config,
                    "test_name": ctx.test_name,
                    "debug_param": ctx.debug_param,
                })
                return ext_info

            return [ScriptNodeData(script_name=self.task.script_name, node=self.task.node, data_group=self.task.data_group, content=item.content if isinstance(item, NodeResult) else item.get('content') if isinstance(item, dict) else None,
                                   ext_info=merge_ext(item.ext_info) if isinstance(item, NodeResult) else merge_ext(item.get('ext_info')) if isinstance(item, dict) else merge_ext(None), source_node=source_node, source_id=source_id, source_nodes=source_nodes, req_id=ctx.req_id, gmt_create=gmt_create,
                                   content_type=item.content_type if isinstance(item, NodeResult) else item.get('content_type') if isinstance(item, dict) else None, content_key=item.content_key if isinstance(item, NodeResult) else item.get('content_key') if isinstance(item, dict) else None) for item
                    in next_datas]

        def __count_task_monitor(self, start_time):
            task = self.task
            monitor_info = self.task.monitor
            monitor_info['last_count_time'] = start_time
            # 计算最近5分钟内的调用次数、成功次数和平均耗时
            ten_minutes_ago = start_time - monitor_info['count_window']
            last_error_detail = None
            for sub_arr in reversed(monitor_info['details']):  # 从后往前遍历
                # 如果子元素的第二个元素小于0 就是异常的
                if sub_arr[1] < 0:
                    last_error_detail = sub_arr
                    break
            if last_error_detail:
                monitor_info['last_error_detail'] = last_error_detail
            monitor_info['details'] = [ts for ts in monitor_info['details'] if ts[0] >= ten_minutes_ago]
            recent_call_count = len(monitor_info['details'])
            success_calls = [tup[1] for tup in monitor_info['details'] if tup[1] > 0]
            recent_success_count = len(success_calls)
            recent_avg_time = sum(success_calls) / recent_call_count if recent_call_count > 0 else 0
            util.log('task_monitor', {
                'task_name': task.name,
                'recent_call_count': recent_call_count,
                'recent_success_count': recent_success_count,
                'recent_avg_time': recent_avg_time
            }, '任务监控更新')
            # 更新
            monitor_info.update({
                'recent_call_count': recent_call_count,
                'recent_success_count': recent_success_count,
                'recent_avg_time': recent_avg_time
            })

            # 如果成功率低于70% 自动停止任务
            if recent_call_count > 0 and recent_success_count / recent_call_count < 0.6:
                util.log('task_break', {
                    'task_name': task.name,
                    'recent_success_rate': recent_success_count / recent_call_count
                }, '成功率低于60%，自动暂停！')
                # 如果第一个就完蛋了不提醒，太敏感了
                if task.success_num > 0:
                    util.tell_dingding({
                        "msgtype": "text",
                        "text": {
                            "content": f'''脚本熔断告警\n脚本：{task.script_name}\n节点：{task.node}\n任务：{task.name}\n最近成功率：{int(recent_success_count * 100 / recent_call_count)}%，低于阈值60%\n自动暂停'''
                        }
                    })
                if task.status == 'on':
                    task.status = 'off'


class EvalTask():
    def __init__(self, name: str = None, script_name: str = None, eval_node: str = None, eval_name: str = None, source: dict = None):
        self.task_type = 'eval_task'
        self.name = name
        self.script_name = script_name
        self.eval_node = eval_node
        self.eval_name = eval_name
        self.status = None
        self.source = source
        self.monitor = {
            'details': [],
            'count_window': 300
        }
        self.queue = queue.Queue()
        self.num = 0
        self.workers = []
        self.success_num = 0
        self.filter_num = 0
        self.fail_num = 0
        self.money = 0

    def start_workers(self, datas: list[script.EvalTaskCtx]):
        self.num = len(datas)
        self.status = 'on'
        if self.num > 0:
            # 初始化队列
            for data in datas:
                if data is not None:
                    self.queue.put(data)

            curr_script = script.get_script_schemas().get(self.script_name)
            if not curr_script:
                raise ValueError('脚本不存在')
            functions = curr_script['functions']

            node_func = functions.get(self.eval_node)
            if not node_func:
                raise ValueError('节点不存在')
            # 一个工作线程
            worker_num = 1
            self.update_workers(worker_num)
            # 等待所有任务完成
            self.queue.join()
        self.clean_on_complete()

    def update_workers(self, worker_num):
        worker_num = worker_num if worker_num > 0 else 1
        current_worker_num = len(self.workers)
        util.log('update_workers', {
            'old_num': current_worker_num,
            'new_num': worker_num,
            'task_name': self.name,
            'script_name': self.script_name,
            'node': self.eval_node
        })
        if current_worker_num < worker_num:
            # 添加新的worker
            for i in range(current_worker_num, worker_num):
                worker = EvalTask.WorkerThread(self, i)
                worker.start()
                self.workers.append(worker)
        elif current_worker_num > worker_num:
            # 停止多余的worker
            for i in range(worker_num, current_worker_num):
                self.workers[i].stop()
            self.workers = self.workers[:worker_num]

    def clean_on_complete(self):
        self.status = 'complete'

    def to_dict(self):
        return {
            "name": self.name,
            'task_type': self.task_type,
            "script_name": self.script_name,
            "eval_node": self.eval_node,
            "eval_name": self.eval_name,
            "worker_num": len(self.workers),
            "num": self.num,
            "success_num": self.success_num,
            "filter_num": self.filter_num,
            "fail_num": self.fail_num,
            "money": self.money,
            "status": self.status,
            "source": self.source,
            "monitor": self.monitor
        }

    def to_json(self):
        return json.dumps(self.to_dict(), ensure_ascii=False, default=str)

    def __repr__(self):
        return self.to_json()

    class WorkerThread(threading.Thread):
        # 工作线程
        def __init__(self, task, index):
            super(EvalTask.WorkerThread, self).__init__()
            self.task = task
            self.index = index
            self.is_stopped = False

        def run(self):
            util.log('eval_worker_start', {
                'task_name': self.task.name,
                'script_name': self.task.script_name,
                'eval_node': self.task.eval_node,
                'worker_index': self.index
            })
            time.sleep(5 + self.index)  # 暂停5秒
            monitor_info = self.task.monitor
            index = 0
            while True:
                if self.is_stopped:
                    util.log('eval_worker_stop', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'eval_node': self.task.eval_node,
                        'worker_index': self.index
                    })
                    break
                try:
                    eval_info = self.task.queue.get()
                except queue.Empty:
                    util.log('eval_queue_empty', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'eval_node': self.task.eval_node,
                        'worker_index': self.index
                    })
                    break
                if not eval_info:
                    continue
                req_id = None
                start_time = time.time()
                try:
                    index += 1
                    if self.task.status == 'delete':
                        util.log('eval_judge_status', {
                            'task_name': self.task.name,
                            'script_name': self.task.script_name,
                            'eval_node': self.task.eval_node,
                            'worker_index': self.index
                        }, '评测删除，跳出执行')
                        break

                    while self.task.status == 'off':
                        util.log('eval_judge_status', {
                            'task_name': self.task.name,
                            'script_name': self.task.script_name,
                            'eval_node': self.task.eval_node,
                            'worker_index': self.index
                        }, '评测暂停，等待执行')
                        time.sleep(10)

                    # 记录开始时间
                    start_time = time.time()
                    eval_name = self.task.eval_name
                    # 重新查一下数据
                    eval_info: script.EvalInfo = eval_data_storage.select_need_eval_datas(filters={
                        'script_name': eval_info.script_name,
                        'source_node': eval_info.source_node,
                        'source_group': eval_info.source_group,
                        'eval_node': eval_info.eval_node,
                        'data_groups': ','.join(eval_info.eval_datas.keys()),
                        'eval_name': eval_name,
                        'source_id': eval_info.source_id
                    })
                    if not eval_info:
                        self.task.filter_num += 1
                        util.log('eval_filter_task', {
                            'task_name': self.task.name,
                            'script_name': self.task.script_name,
                            'eval_node': self.task.eval_node,
                            'do_index': str(index),
                            'worker_index': self.index
                        }, '评测跳过数据')
                        continue
                    #     前面最多查取第一个
                    eval_info = eval_info[0]
                    req_id = str(time.time()) + '.' + str(self.index)
                    util.log('eval_start_task', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'eval_node': self.task.eval_node,
                        'do_index': str(index),
                        'worker_index': self.index,
                        'info': eval_info
                    }, '处理数据')

                    # script_thread_local.run_node_data = {
                    #     'task': self.task,
                    #     'script_name': self.task.script_name,
                    #     'node': self.task.node,
                    #     'data_group': self.task.data_group,
                    #     'worker_num': len(self.task.workers),
                    #     'data': data,
                    #     'req_id': req_id,
                    #     'test_name': eval_name,
                    #     'task_config': task_config
                    # }
                    util.log_local.run_data = {
                        'run_task_id': req_id,
                        'eval_name': eval_name
                    }
                    ctx = script.EvalTaskCtx(task_name=self.task.name, eval_name=self.task.eval_name, req_id=req_id)
                    eval_res = self.run_eval(eval_info, ctx)

                    common_storage.eval_data_storage.add_eval_data(eval_res)
                    self.task.success_num += 1
                    cost = math.ceil((time.time() - start_time) * 1000)
                    monitor_info['details'].append((start_time, cost, None, req_id))
                    # 10s统计一次
                    if monitor_info.get('last_count_time') is None or start_time - monitor_info.get('last_count_time') >= 10:
                        self.__count_task_monitor(start_time)
                except Exception as e:
                    self.task.fail_num += 1
                    monitor_info['details'].append((start_time, -1, traceback.format_exc(), req_id))
                    # 一旦失败就计算
                    self.__count_task_monitor(start_time)
                    util.log('eval_task_error', {
                        'task_name': self.task.name,
                        'script_name': self.task.script_name,
                        'eval_node': self.task.eval_node,
                        'data': eval_info,
                        'error': e
                    }, '评测节点处理异常')
                finally:
                    # 清理线程局部变量
                    # if hasattr(script_thread_local, "run_node_data"):
                    #     del script_thread_local.run_node_data
                    # 清理线程局部变量
                    if hasattr(util.log_local, "run_data"):
                        del util.log_local.run_data
                    self.task.queue.task_done()

        def stop(self):
            self.is_stopped = True

        def run_eval(self, data: script.EvalInfo, ctx: script.EvalTaskCtx = None):
            # script_name 是py脚本名称，该脚本在scripts目录下
            # node 是py脚本下的函数名称 该node有注解snode
            # 根据 script_name 和 node 获取指定函数，并将data传入 执行他，并将它的返回结果返回

            schema = script.get_script_schemas().get(data.script_name)
            if not schema:
                #     脚本不存在，可能是被删除了或者改名了。
                raise ValueError('脚本不存在，可能被删除了')
            module = script.get_script_modules().get(data.script_name)
            if not module:
                #     脚本不存在，可能是被删除了或者改名了。
                raise ValueError('脚本不存在，可能被删除了')
            # Get the function from the module
            functions = schema['functions']
            node_func = functions.get(data.eval_node)
            if not node_func:
                raise ValueError('节点不存在')
            eval_func = node_func.get('eval_func')
            if not eval_func:
                raise ValueError('节点未配置评测函数')
            eval_result = getattr(module, eval_func)(data, ctx)
            eval_results = {}
            eval_ext_info = {}
            if not eval_result:
                pass
            elif isinstance(eval_result, dict):
                eval_results = eval_result.get('results')
                eval_ext_info = eval_result.get('ext_info')
            else:
                eval_results = eval_result.results
                eval_ext_info = eval_result.ext_info

            return ScriptEvalData(script_name=data.script_name, source_node=data.source_node, source_group=data.source_group, eval_name=data.eval_name, source_id=data.source_id, source_content_type=data.source_content_type,source_content=data.source_content, eval_node=data.eval_node, req_id=ctx.req_id, eval_datas=data.eval_datas,
                                  results=eval_results, ext_info=eval_ext_info)

        def __count_task_monitor(self, start_time):
            task = self.task
            monitor_info = self.task.monitor
            monitor_info['last_count_time'] = start_time
            # 计算最近5分钟内的调用次数、成功次数和平均耗时
            ten_minutes_ago = start_time - monitor_info['count_window']
            last_error_detail = None
            for sub_arr in reversed(monitor_info['details']):  # 从后往前遍历
                # 如果子元素的第二个元素小于0 就是异常的
                if sub_arr[1] < 0:
                    last_error_detail = sub_arr
                    break
            if last_error_detail:
                monitor_info['last_error_detail'] = last_error_detail
            monitor_info['details'] = [ts for ts in monitor_info['details'] if ts[0] >= ten_minutes_ago]
            recent_call_count = len(monitor_info['details'])
            success_calls = [tup[1] for tup in monitor_info['details'] if tup[1] > 0]
            recent_success_count = len(success_calls)
            recent_avg_time = sum(success_calls) / recent_call_count if recent_call_count > 0 else 0
            util.log('eval_task_monitor', {
                'task_name': task.name,
                'recent_call_count': recent_call_count,
                'recent_success_count': recent_success_count,
                'recent_avg_time': recent_avg_time
            }, '评测任务监控更新')
            # 更新
            monitor_info.update({
                'recent_call_count': recent_call_count,
                'recent_success_count': recent_success_count,
                'recent_avg_time': recent_avg_time
            })

            # 如果成功率低于70% 自动停止任务
            if recent_call_count > 0 and recent_success_count / recent_call_count < 0.6:
                util.log('eval_task_break', {
                    'task_name': task.name,
                    'recent_success_rate': recent_success_count / recent_call_count
                }, '成功率低于60%，自动暂停！')
                # 如果第一个就完蛋了不提醒，太敏感了
                if task.success_num > 0:
                    util.tell_dingding({
                        "msgtype": "text",
                        "text": {
                            "content": f'''评测脚本熔断告警\n脚本：{task.script_name}\n评测节点：{task.eval_node}\n任务：{task.name}\n最近成功率：{int(recent_success_count * 100 / recent_call_count)}%，低于阈值60%\n自动暂停'''
                        }
                    })
                if task.status == 'on':
                    task.status = 'off'


def get_all_subclasses(cls):
    all_subclasses = {}
    for subclass in cls.__subclasses__():
        all_subclasses[subclass.__name__] = subclass
        all_subclasses.update(get_all_subclasses(subclass))
    return all_subclasses


one_runtime.selects = get_all_subclasses(DataSelector)
